package cluster

import (
	"cluster-cache/meta"
	"cluster-cache/rpc"
	"cluster-cache/store"
	"encoding/binary"
	"fmt"
	"github.com/cespare/xxhash/v2"
	"github.com/hashicorp/raft"
	"github.com/panjf2000/gnet/pkg/logging"
	"io"
	"net"
	"os"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

const raftTimeout = 10 * time.Second

type SingleFsm struct {
	bootID     int
	store      *store.MemoryStore
	logStore   *store.MMapLog
	Raft       *raft.Raft
	leaderCh   chan raft.Observation
	leaderLock *sync.Mutex

	applyCond *sync.Cond
	applyLock *sync.RWMutex
	applied   *uint64

	rpc     *rpc.TcpTransport
	rpcUrls []string

	Start bool
}

const RaftPort = 6060
const RpcPort = 1000
const ApplyTimeOut = time.Second

func NewSingleFsm(store *store.MemoryStore, hosts []string, bootID int, masterId int, dir string, config *Config) *SingleFsm {
	lock := new(sync.RWMutex)
	fsm := &SingleFsm{
		bootID:     bootID,
		store:      store,
		leaderCh:   make(chan raft.Observation),
		leaderLock: new(sync.Mutex),
		applyCond:  sync.NewCond(lock),
		applyLock:  lock,
		applied:    new(uint64),
		Start:      false,
	}
	urls := buildUrlFromHost(hosts, RaftPort)
	fsm.rpcUrls = buildUrlFromHost(urls, RpcPort)
	logging.Infof("start rpc %s in %v, boot=%d", fsm.rpcUrls[bootID], fsm.rpcUrls, bootID)
	rpcTrans := rpc.NewTcpTransport(fsm.rpcUrls, bootID, fsm)
	fsm.rpc = rpcTrans
	logging.Infof("start raft %s in %v, boot=%d", urls[bootID], urls, bootID)
	tcpAddr, err := net.ResolveTCPAddr("tcp", urls[bootID])
	if err != nil {
		logging.Fatalf("resolve tcp error: %s", err)
		return nil
	}
	transport, err := raft.NewTCPTransport(urls[bootID], tcpAddr, 10, 10*time.Second, os.Stderr)
	if err != nil {
		logging.Fatalf("tcp transport error: %s", err)
		return nil
	}
	name := fmt.Sprintf("%s/fsm-%d", dir, bootID)
	fsm.Raft, fsm.logStore = NewRaft(bootID, fsm, name, transport)
	bootstrap(fsm.Raft, strconv.Itoa(bootID), transport, bootID == masterId, urls, config.NeedInit())
	// reading leader change
	observer := raft.NewObserver(fsm.leaderCh, false, seeNewLeader)
	fsm.Raft.RegisterObserver(observer)
	// save config
	err = config.SaveCluster(urls, bootID)
	if err != nil {
		logging.Fatalf("save cluster data error: %s", err)
		return nil
	}
	return fsm
}

func buildUrlFromHost(hosts []string, delta int) []string {
	urls := make([]string, 0, len(hosts))
	for i := 0; i < len(hosts); i++ {
		hostPort := strings.Split(hosts[i], ":")
		from := 0
		if len(hostPort) > 1 {
			port, err := strconv.Atoi(hostPort[1])
			if err != nil {
				panic(err)
			}
			from = port
			if delta == RaftPort {
				delta = 0
			}
		}
		urls = append(urls, hostPort[0]+":"+strconv.Itoa(from+delta))
	}
	return urls
}

// Accept 处理master从slaver接受的数据
func (s *SingleFsm) Accept(data []byte) []rpc.CallFuture {
	id := binary.LittleEndian.Uint16(data[0:])
	if id == meta.UpdateIndexFromMaster {
		index := binary.LittleEndian.Uint64(data[2:])
		hash := binary.LittleEndian.Uint64(data[10:])
		s.store.Indexes.Update(hash, index)
		return nil
	} else if id == meta.InitClusterSuccess {
		s.Start = true
		return nil
	}
	copied := make([]byte, len(data))
	copy(copied, data)
	return s.Write(copied, ApplyTimeOut)
}

func (s *SingleFsm) Apply(raftLog *raft.Log) interface{} {
	if raftLog == nil || raftLog.Data == nil {
		return nil
	}
	pos := 0
	data := raftLog.Data
	size := len(data)
	for pos < size {
		kvs, m := meta.Decode(data)
		if kvs == nil {
			break
		}
		s.store.Apply(raftLog.Index, s.bootID, kvs)
		pos += m
		data = data[m:]
	}
	s.PublishIndex(raftLog.Index)
	return nil
}

func (s *SingleFsm) PublishIndex(index uint64) {
	s.applyLock.Lock()
	atomic.StoreUint64(s.applied, index)
	s.applyLock.Unlock()
	s.applyCond.Broadcast()
}

func (s *SingleFsm) WaitApply(index uint64) {
	s.applyCond.L.Lock()
	for index > atomic.LoadUint64(s.applied) {
		s.applyCond.Wait()
	}
	s.applyCond.L.Unlock()
}

func (s *SingleFsm) WaitFlush(index uint64) {
	s.logStore.WaitFlush(index)
}

func (s *SingleFsm) LastIndex(hash uint64) uint64 {
	return s.store.Indexes.Get(hash)
}

func (s *SingleFsm) Snapshot() (raft.FSMSnapshot, error) {
	return nil, nil
}

func (s *SingleFsm) Restore(snapshot io.ReadCloser) error {
	return nil
}

func (s *SingleFsm) IsLeader() bool {
	return s.Raft.State() == raft.Leader
}

// Write 写数据
func (s *SingleFsm) Write(cmd []byte, timeout time.Duration) []rpc.CallFuture {
	if s.IsLeader() {
		// leader 异步写数据, 然后并批量所有节点index
		apply := s.Raft.Apply(cmd, timeout)
		_ = apply.Error()
		index := apply.Index()
		// cmd中可能包含多个指令, 需要依次解码
		futures := make([]rpc.CallFuture, 0)
		pos := 0
		data := cmd
		size := len(data)
		for pos < size {
			kvs, m := meta.Decode(data)
			if kvs == nil {
				break
			}
			// 发布hash index
			hash := xxhash.Sum64(kvs.Key)
			buf := meta.BuildUpdateIndexFromMaster(index, hash)
			future := s.Notice(buf)
			futures = append(futures, future...)
			pos += m
			data = data[m:]
		}
		flushFuture := &rpc.FlushFuture{
			Index:    index,
			LogStore: s.logStore,
		}
		return append(futures, flushFuture)
	} else {
		// slaver 发送到master, 等待完成
		_, id := s.GetLeader()
		i, _ := strconv.Atoi(string(id))
		conn := s.rpc.Client.GetConn(s.rpcUrls[i])
		future := conn.Send(cmd)
		return append(make([]rpc.CallFuture, 0, 1), future)
	}
}

func (s *SingleFsm) Notice(buf []byte) []rpc.CallFuture {
	futures := make([]rpc.CallFuture, 0, 3)
	for i := 0; i < len(s.rpcUrls); i++ {
		if i == s.bootID {
			continue
		}
		conn := s.rpc.Client.GetConn(s.rpcUrls[i])
		future := conn.Send(buf)
		if future != nil {
			futures = append(futures, future)
		}
	}
	return futures
}

func seeNewLeader(o *raft.Observation) bool { _, ok := o.Data.(raft.LeaderObservation); return ok }

func (s *SingleFsm) GetLeader() (raft.ServerAddress, raft.ServerID) {
	leader, id := s.Raft.LeaderWithID()
	//log.Printf("leader=%s", leader)
	for len(leader) == 0 {
		s.leaderLock.Lock()
		leader, id = s.Raft.LeaderWithID()
		if len(leader) != 0 {
			s.leaderLock.Unlock()
			break
		}
		<-s.leaderCh
		leader, id = s.Raft.LeaderWithID()
		//log.Printf("leader=%s", leader)
		s.leaderLock.Unlock()
	}
	return leader, id
}
